#include "base/base_buffer.h"
#include "base/channel.h"
#include "util/count_down_cond.h"
#include "util/cpuinfo.h"
#include "base/eventloop.h"
#include "base/eventloopthreadpool.h"
#include "base/logthread.h"
#include "base/timer.h"
#include "base/thread.h"
#include "base/threadpool.h"
#include "net/accept.h"
#include "net/buffer.h"
#include "net/connect.h"
#include "net/connectpool.h"

#include <signal.h>
#include <sys/time.h>
#include <sys/syscall.h>
#include <unistd.h>
#include <fcntl.h>

#include "net/tcpserver.h"
namespace net_tools
{
  // void Sigpipe(int sig){
  //   NT_LOG_INFO << "BROKEN PIPE" << NT_LOG_ENDL;
  // }
  std::atomic<int> Tcpserver::connect_number_;

  Tcpserver::Tcpserver()
      : count_down_cond_(new util::Count_down_cond(-1)),
        eventloop_(new Eventloop(_C::CHANNEL_DEFAULT_SIZE, base::Thread::thread_tid_)),
        accept_(new net::Accept([&]
                                { New_connect_call_back(); },
                                eventloop_)), //lambda
        accept_fd(accept_->Get_accept_fd()),
        sockaddr_in_size(sizeof(sockaddr_in)),

        cpu_high_level_callback_([]
                                 { ; }),
        before_loop_todo_([](Eventloop *eventloop)
                          { ; }),

        connect_over_callback_([](net::Connect *connect)
                               { connect->Print_ip_port("connect"); }),
        message_callback_(Default_message),
        close_callback_(Default_close),
        lose_callback_(Default_lose)
  {
    connect_number_.store(0);
    signal(SIGPIPE, SIG_IGN);
    // sgiaction版本
    // struct sigaction sa;
    // sa.sa_handler = Sigpipe;
    // sigemptyset(&sa.sa_mask);
    // sa.sa_flags = 0;
    // sigaction(SIGPIPE,&sa,NULL);
  };
  void Tcpserver::Initserver(int thread_number, int work_thread_number, int mempool_block_number, int mempool_block_size, std::string name)
  {
    net::Connect::threadpool_ = new net_tools::base::Threadpool(work_thread_number);
    net_tools::base::Base_buffer::Singleton_pool(mempool_block_size, mempool_block_number);
    thread_todo = [&](Eventloop *eventloop)
    { Default_thread(eventloop, before_loop_todo_, lose_callback_, count_down_cond_); };
    eventloopthreadpool_ = new Eventloopthreadpool(thread_todo, thread_number);
    eventlooppool_ = eventloopthreadpool_->Get_ptr_looppool();
    thread_number_ = eventloopthreadpool_->Get_thread_number();
    count_down_cond_->Add_count(thread_number_ + 1);
    count_down_cond_->Wait();
    delete count_down_cond_;
    base::Thread::p_thread_name_ = new std::string(name);
  };
  void Tcpserver::Startserver(int port)
  {
    accept_->Init_listen(port);
    eventloop_->Loop();
  };
  int Tcpserver::Get_connect_number()
  {
    int connect_number = 0;
    for (int count = 0; count < thread_number_; ++count)
    {
      connect_number += (*eventlooppool_)[count]->connectpool_->Get_connect_number();
    }
    return connect_number;
  };
  void Tcpserver::Add_timejob(const std::string job_name, func todo, timeval begin_time, timeval time, int count)
  {
    eventloop_->Run_every(std::move(todo), begin_time, time, count, std::move(job_name));
  };
  void Tcpserver::Add_timejob_after_sec(const std::string job_name, func todo, long int begin_time, long int time, int count)
  {
    timeval now_, sec_;
    gettimeofday(&now_, NULL);
    now_.tv_sec += begin_time;
    sec_.tv_sec = time;
    sec_.tv_usec = 0;
    eventloop_->Run_every(std::move(todo), now_, sec_, count, std::move(job_name));
  };
  void Tcpserver::Del_timejob(const std::string job_name)
  {
    eventloop_->Deltimejob_every(std::move(job_name));
  };
  net_tools::base::Timeevent *Tcpserver::Add_once_timejob_at(func func_, timeval begin_time)
  {
    return eventloop_->Run_at(std::move(func_), begin_time);
  };
  net_tools::base::Timeevent *Tcpserver::Add_once_timejob_after(func func_, timeval begin_time)
  {
    return eventloop_->Run_after(std::move(func_), begin_time);
  };
  net_tools::base::Timeevent *Tcpserver::Add_once_timejob_after_sec(func func_, long int sec)
  {
    timeval now_;
    gettimeofday(&now_, NULL);
    now_.tv_sec += sec;
    return eventloop_->Run_at(std::move(func_), now_);
  };
  bool Tcpserver::Del_once_timejob(net_tools::base::Timeevent *timeevent)
  {
    return eventloop_->Deltimejob_once(timeevent);
  };

  void Tcpserver::New_connect(Eventloop *eventloop, sockaddr_in new_connect_addr, int new_connect_fd)
  {
    timeval now;
    gettimeofday(&now, NULL);
    base::Channel *channel = eventloop->Get_free_channel(new_connect_fd);
    net::Connect *connect = eventloop->connectpool_->Get_Connect(channel);
    connect->Set_address(new_connect_addr);
    connect->connect_time_ = now;
    //connect初始化高效 未完成
    connect->buffer_.Set_able_read_callback(message_callback_);
    connect->buffer_.Set_able_close_callback(close_callback_);
    connect->buffer_.Set_able_user_close_callback(Default_user_close);
    eventloop->connectpool_->Add_heartbeat(connect);
    channel->Enableread(); //Enable recv data
    connect_over_callback_(connect);
    channel->Add_to_epoll();
  };
  void Tcpserver::New_connect_call_back()
  {
    struct sockaddr_in accept_addr;
    while (true)
    {
      int new_connect_fd = accept4(accept_fd, reinterpret_cast<sockaddr *>(&accept_addr), &sockaddr_in_size, SOCK_NONBLOCK);
      if (new_connect_fd != -1) [[likely]]
      { //when no connect,return -1
        // if(connect_number_.load() > _C::MAX_CONNECT_NUMBER) [[unlikely]] {
        //   NT_LOG_WARN << "OVER MAX CONNECT NUMBER" << NT_LOG_ENDL;
        //   close(new_connect_fd);
        //   return;
        // }else if(base::Cpuinfo::cpu_loadavg_ >= _C::HIGH_CPU_LOADAVG){
        //   NT_LOG_WARN << "OVER MAX CONNECT NUMBER" << NT_LOG_ENDL;
        //   close(new_connect_fd);
        //   cpu_high_level_callback_();
        //   return;
        // }
        net::Socket::Set_noblocked(new_connect_fd);
        connect_number_++;
        Eventloop *next_eventloop = eventloopthreadpool_->Get_next_eventloop();
        next_eventloop->Run_in_thread([this, accept_addr, new_connect_fd, next_eventloop]
                                      { New_connect(next_eventloop, accept_addr, new_connect_fd); });
      }
      else
      {
        return;
      }
    }
  };

  void Tcpserver::Default_thread(Eventloop *eventloop, eventloop_func before_loop_todo, connect_func lose_callback, util::Count_down_cond *count_down_cond)
  {
    eventloop->Init_thread_id();
    net::Connectpool *connectpool = new net::Connectpool(eventloop, 1024);
    eventloop->connectpool_ = connectpool;
    eventloop->connectpool_->Set_heartbeat(std::move(lose_callback));
    eventloop->connectpool_->Start_heartbeat();
    count_down_cond->Signal(); //初始化完成
    before_loop_todo(eventloop);
    eventloop->Loop();
  };

  void Tcpserver::Default_message(net::Connect *connect)
  {
    connect->Read_callback();
  };

  void Tcpserver::Default_close(net::Connect *connect)
  {
    connect->Heartbeat_Close_connect_todo();
    connect->Close_callback();
    connect_number_--;
  };
  void Tcpserver::Default_lose(net::Connect *connect)
  {
    connect->Lose_heartbeat_callback();
    connect_number_--;
  };

  void Tcpserver::Default_user_close(net::Connect *connect)
  {
    connect->Heartbeat_Close_connect_todo();
    connect->User_want_close_callback();
    connect_number_--;
  };

} // namespace net_tools